package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.UserRegisterBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsUserUserRegisterWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsUserUserRegisterWindow().start(Constant.TOPIC_DWD_USER_REGISTER,Constant.DWS_USER_USER_REGISTER_WINDOW,4,10025);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        // 1.读取数据源
//        dataStreamSource.print();

        // 2.etl

        SingleOutputStreamOperator<JSONObject> streamOperator = eltStream(dataStreamSource);

        SingleOutputStreamOperator<UserRegisterBean> streamOperator1 = rgStream(streamOperator);
//        process.print();

        // 3. 添加水位线
        SingleOutputStreamOperator<UserRegisterBean> waterStream = WaterStream(streamOperator1);
//        waterStream.print();

        //开窗
        SingleOutputStreamOperator<UserRegisterBean> reduceStream = reduceStream(waterStream);

        // 4.写到doris
        reduceStream.print();

        reduceStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_USER_USER_REGISTER_WINDOW));

    }

    private static SingleOutputStreamOperator<UserRegisterBean> reduceStream(SingleOutputStreamOperator<UserRegisterBean> waterStream) {
         return waterStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(1)))
                .reduce(new ReduceFunction<UserRegisterBean>() {
                    @Override
                    public UserRegisterBean reduce(UserRegisterBean t1, UserRegisterBean t2) throws Exception {
                        t1.setRegisterCt(t1.getRegisterCt() + t2.getRegisterCt());
                        return t1;
                    }
                }, new AllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<UserRegisterBean> iterable, Collector<UserRegisterBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<UserRegisterBean> iterator = iterable.iterator();

                        UserRegisterBean userRegisterBean = iterator.next();

                        userRegisterBean.setStt(DateFormatUtil.tsToDateTime(start));
                        userRegisterBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        userRegisterBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));

                        collector.collect(userRegisterBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<UserRegisterBean> WaterStream(SingleOutputStreamOperator<UserRegisterBean> streamOperator1) {
        return streamOperator1.assignTimestampsAndWatermarks(WatermarkStrategy
                .<UserRegisterBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getCreate_time()).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<UserRegisterBean> rgStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
         return streamOperator.process(new ProcessFunction<JSONObject, UserRegisterBean>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, UserRegisterBean>.Context context, Collector<UserRegisterBean> collector) throws Exception {

                Long createtime = jsonObject.getLong("create_time");

                String id = jsonObject.getString("id");

                collector.collect(UserRegisterBean.builder()
                        .registerCt(1L)
                        .curDate(DateFormatUtil.tsToDate(createtime))
                        .create_time(createtime)
                        .build());
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject != null) {
                        String id = jsonObject.getString("id");
                        String createTime = jsonObject.getString("create_time");

                        if (id != null && createTime != null) {
                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
